package com.gitee.ixtf.vertx.rabbitmq.verticle

import com.gitee.ixtf.guice.Jguice
import com.gitee.ixtf.vertx.verticle.BaseCoroutineVerticle
import io.vertx.core.Future
import io.vertx.kotlin.coroutines.coAwait
import io.vertx.rabbitmq.RabbitMQClient
import io.vertx.rabbitmq.RabbitMQMessage
import io.vertx.rabbitmq.RabbitMQOptions
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.isActive

abstract class RabbitMQVerticle : BaseCoroutineVerticle() {
  private val options: RabbitMQOptions by Jguice
  protected val client: RabbitMQClient by lazy {
    RabbitMQClient.create(vertx, options.also { prepare(it) })
  }
  protected open val prefetchCount = 1

  protected open fun prepareClient(): Future<Void> = client.basicQos(prefetchCount)

  override suspend fun start() {
    super.start()
    client.addConnectionEstablishedCallback { prepareClient().onComplete(it) }
    client.start().coAwait()
  }

  override suspend fun stop() {
    client.stop().coAwait()
    super.stop()
  }

  protected open fun prepare(options: RabbitMQOptions) {
    options.connectionName = this::class.java.name
  }

  protected suspend fun RabbitMQMessage.basicAck(multiple: Boolean = false) {
    client.basicAck(envelope().deliveryTag, multiple).coAwait()
  }

  protected suspend fun RabbitMQMessage.basicNack(
      t: Throwable,
      multiple: Boolean = false,
      requeue: Boolean = false
  ) {
    if (!isActive || t is CancellationException) return
    log.error(t, "${envelope()}")
    client.basicNack(envelope().deliveryTag, multiple, requeue).coAwait()
  }
}
